package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import com.sfx.mq.constants.TopicProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-28
 * Time: 9:26
 */
public class TagsMessageTest {


    @Test
    public void producer() throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("producer-group");
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        Message message1 = new Message(TopicProperties.TAG_TOPIC,"vip1", "vip1消息".getBytes());
        Message message2 = new Message(TopicProperties.TAG_TOPIC,"vip2", "vip2消息".getBytes());
        producer.send(message1);
        producer.send(message2);
        System.out.println("发送消息成功");
        // 关闭生产者实例
        producer.shutdown();
    }

    @Test
    public void consumer_a() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-a");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);

        consumer.subscribe(TopicProperties.TAG_TOPIC,"vip1");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费vip1的消费者");
                String msg = new String(list.get(0).getBody());
                System.out.println("消费的消息为 : " + msg);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        // 悬停,等待消费完成
        System.in.read();
        consumer.shutdown();
    }

    @Test
    public void consumer_b() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-b");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);

        consumer.subscribe(TopicProperties.TAG_TOPIC,"vip1 || vip2");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费vip1 || vip2 的消费者");
                String msg = new String(list.get(0).getBody());
                System.out.println("消费的消息为 : " + msg);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        // 悬停,等待消费完成
        System.in.read();
        consumer.shutdown();
    }
}